package core.rdd.血缘;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import scala.Tuple2;

import java.util.Arrays;
import java.util.List;

public class Spark01_xueyuan {

    public static void main(String[] args) {
        /**
         *flatMap 是将多集合转换成一个
         */
        // 配置SparkConf指向你的Spark master URL
        SparkConf conf = new SparkConf()
                .setAppName("Spark01_xueyuan") // 应用名称
                .setMaster("local[*]"); // 替换成你的master地址
        JavaSparkContext sc = new JavaSparkContext(conf);
        // 创建JavaSparkContext，它是与集群交互的主要入口点
        try {
            Tuple2<String, Integer> tuple1 = new Tuple2<>("a", 123);
            Tuple2<String, Integer> tuple2 = new Tuple2<>("b", 456);
            Tuple2<String, Integer> tuple3 = new Tuple2<>("c", 5);
            Tuple2<String, Integer> tuple4 = new Tuple2<>("d", 6);
            Tuple2<String, Integer> tuple5 = new Tuple2<>("e", 2);
            Tuple2<String, Integer> tuple6 = new Tuple2<>("f", 4);


            List<Tuple2<String, Integer>> tuple2s = Arrays.asList(
                    tuple1, tuple2, tuple3, tuple5, tuple4, tuple6
            );

            JavaPairRDD<String, Integer> stringIntegerJavaPairRDD = sc.parallelizePairs(tuple2s, 2);
            System.out.println("stringIntegerJavaPairRDD的血缘:" + stringIntegerJavaPairRDD.toDebugString());

            JavaPairRDD<String, Object> stringObjectJavaPairRDD = stringIntegerJavaPairRDD.mapValues(new Function<Integer, Object>() {
                @Override
                public Object call(Integer v1) throws Exception {
                    return v1 * 2;
                }
            });

            System.out.println("################################################################################################");
            System.out.println("stringObjectJavaPairRDD的血缘:" + stringObjectJavaPairRDD.toDebugString());



        } finally {
            sc.close();
        }
    }
}
